package org.topteam.oschat.akka;

import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;

import org.topteam.oschat.entity.User;
import org.topteam.oschat.event.SessionEvent;
import org.topteam.oschat.event.SessionEvent.SessionClose;
import org.topteam.oschat.event.SessionEvent.SessionOpen;
import org.topteam.oschat.message.TextMessage;

import akka.actor.ActorRef;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.cluster.routing.AdaptiveLoadBalancingGroup;
import akka.cluster.routing.ClusterRouterGroup;
import akka.cluster.routing.ClusterRouterGroupSettings;
import akka.cluster.routing.HeapMetricsSelector;

public class SessionActor extends UntypedActor {

	private Map<String, ActorRef> users = new HashMap<>();

	public SessionActor() {
	}

	@Override
	public void onReceive(Object message) throws Exception {
		if (message instanceof SessionEvent.SessionOpen) {
			SessionOpen open = (SessionOpen) message;
			User user = open.getUser();
			ActorRef userAc = users.get(user.getUserId());
			if (userAc == null) {

				userAc = getContext().actorOf(
						Props.create(UserActor.class, user), user.getUserId());
				userAc.tell(open, getSelf());
				users.put(user.getUserId(), userAc);
				getContext().watch(userAc);

			} else {
				userAc.tell(open, userAc);
			}
		} else if (message instanceof SessionEvent.SessionClose) {
			SessionClose close = (SessionClose) message;
			getRemoteActor("session/" + close.getUserId()).tell(close,
					getSelf());
		} else if (message instanceof TextMessage) {
			TextMessage msg = (TextMessage) message;
			getRemoteActor("session/" + msg.getTo()).tell(msg, getSelf());
		}
	}

	private ActorRef getRemoteActor(String id) {
		int totalInstances = 100;
		Iterable<String> routeesPaths = Arrays.asList("/user/" + id);
		boolean allowLocalRoutees = true;
		ClusterRouterGroup clusterRouterGroup = new ClusterRouterGroup(
				new AdaptiveLoadBalancingGroup(
						HeapMetricsSelector.getInstance(),
						Collections.<String> emptyList()),
				new ClusterRouterGroupSettings(totalInstances, routeesPaths,
						allowLocalRoutees, "oschat"));
		ActorRef remoteActor = getContext().system().actorOf(
				clusterRouterGroup.props());
		return remoteActor;
	}
}
